package com.atguigu.flink.cep;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/11/26
 */
public class Demo7_Option
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
                    .<WaterSensor>forMonotonousTimestamps()
                    .withTimestampAssigner((e, r) -> e.getTs());


                SingleOutputStreamOperator<WaterSensor> ds = env.readTextFile("data/cep.txt")
                                                                .map(new MapFunction<String, WaterSensor>()
                                                               {
                                                                   @Override
                                                                   public WaterSensor map(String value) throws Exception {
                                                                       String[] words = value.split(",");
                                                                       return new WaterSensor(words[0], Long.valueOf(words[1]), Integer.valueOf(words[2]));
                                                                   }
                                                               })
                                                                .assignTimestampsAndWatermarks(watermarkStrategy);


                Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("规则1")
                    .where(new SimpleCondition<WaterSensor>()
                    {
                        //定义匹配规则
                        @Override
                        public boolean filter(WaterSensor value) throws Exception {
                            return "s1".equals(value.getId());
                        }
                    })
                    .optional() // 可选，规则1可有可无
                    // 有：  s1 next s2
                    // 无:   s2
                    .next("s2")
                    .where(new SimpleCondition<WaterSensor>()
                    {
                        //定义匹配规则
                        @Override
                        public boolean filter(WaterSensor value) throws Exception {
                            return "s2".equals(value.getId());
                        }
                    })
                    ;


                PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);


                patternStream.select(new PatternSelectFunction<WaterSensor, String>()
                {

                    @Override
                    public String select(Map<String, List<WaterSensor>> map) throws Exception {
                        return map.toString();
                    }
                })
                .print().setParallelism(1);


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
